package com.youzi.tunnel.server.handlers;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.concurrent.ExecutionException;

import com.alibaba.fastjson.JSONObject;
import com.youzi.tunnel.common.protocol.Constants;
import com.youzi.tunnel.common.protocol.Message;
import com.youzi.tunnel.server.config.ServerProperties;
import com.youzi.tunnel.server.entity.Client;
import com.youzi.tunnel.server.entity.Tunnel;
import com.youzi.tunnel.server.manager.ChannelManager;
import com.youzi.tunnel.server.manager.FutureManager;
import com.youzi.tunnel.server.manager.TunnelManager;
import com.youzi.tunnel.server.manager.ClientManager;
import com.youzi.tunnel.server.work.UserClientBootstrap;
import com.youzi.tunnel.server.work.UserServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import lombok.extern.slf4j.Slf4j;

import io.netty.buffer.Unpooled;

/**
 * 服务端处理器
 */
@Slf4j
public class ServerChannelHandler extends SimpleChannelInboundHandler<Message> {
    private static final boolean isDebugger = ServerProperties.getInstance().isDebugger();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message message) throws Exception {
        switch (message.getType()) {
            case HEARTBEAT:
                handleHeartbeatMessage(ctx, message);
                break;
            case LINK:
                handleLinkMessage(ctx, message);
                break;
            case OPEN_CONNECT:
                handleOpenConnectMessage(ctx, message);
                break;
            case CLOSE_CONNECT:
                handleCloseConnectMessage(ctx, message);
                break;
            case OPEN_PORT:
                handleOpenPortMessage(ctx, message);
                break;
            case CLOSE_PORT:
                handleClosePortMessage(ctx, message);
                break;
            case UNLINK:
                handleUnlinkMessage(ctx, message);
                break;
            case RELAY:
                handleTransferMessage(ctx, message);
                break;
            default:
                break;
        }
    }

    private void handleTransferMessage(ChannelHandlerContext ctx, Message message) {
        Channel channel = ctx.channel();
        Channel toChannel = channel.attr(Constants.TOWARD_CHANNEL).get();
        InetSocketAddress address = (InetSocketAddress) (ctx.channel().remoteAddress());

        assert toChannel != null;
        Boolean channelOnServer = toChannel.attr(Constants.CHANNEL_ON_SERVER).get();
        if (channelOnServer != null && channelOnServer) {
            //转发到服务端监听端口
            byte[] data = message.getData();
            if (isDebugger) {
                log.info("主机[{}:{}]请求转发数据到服务器目标端口, 数据长度:{}bytes", address.getHostString(), address.getPort(), data.length);
            }

            int leaveSize = data.length;
            int buffSize = 1024;
            int index = 0;

            while (leaveSize > 0) {
                int sendSize = Math.min(leaveSize, buffSize);
                ByteBuf buf = ctx.alloc().buffer(sendSize);

                buf.writeBytes(data, index, sendSize);

                //转发消息
                toChannel.writeAndFlush(buf);
                leaveSize -= sendSize;
                index += sendSize;
            }

        } else {
            //转发到客户端
            if (isDebugger) {
                log.info("主机[{}:{}]请求转发数据到目标客户端,数据长度:{}bytes", address.getHostString(), address.getPort(), message.getData().length);
            }
            //转发消息
            toChannel.writeAndFlush(message);
        }

    }

    private void handleUnlinkMessage(ChannelHandlerContext ctx, Message message) throws InterruptedException {
        String clientId = message.getContent();
        //修改隧道信息
        Collection<Tunnel> byFromClient = TunnelManager.getByFromClient(clientId);

        byFromClient.forEach(e -> {
            Channel channel = ClientManager.getClient(e.getFromClient()).getChannel();
            //下发断开信息  TODO
            channel.writeAndFlush(Unpooled.EMPTY_BUFFER);

            //修改状态为关闭
            e.setAlive(false);
        });
        Collection<Tunnel> byToClient = TunnelManager.getByToClient(clientId);
        byToClient.forEach(e -> {
            Channel channel = ClientManager.getClient(e.getToClient()).getChannel();
            //下发断开信息  TODO
            channel.writeAndFlush(Unpooled.EMPTY_BUFFER);

            //修改状态为关闭
            e.setAlive(false);
        });
        //修改客户端信息
        Client client = ClientManager.getClient(clientId);
        client.setChannel(null);
        client.setOnline(false);
        //下发断开信息  TODO
        ctx.channel().writeAndFlush(Unpooled.EMPTY_BUFFER);
        ctx.channel().close().sync();
    }

    private void handleLinkMessage(ChannelHandlerContext ctx, Message message) throws ExecutionException, InterruptedException {
        Channel channel = ctx.channel();
        InetSocketAddress address = (InetSocketAddress) (ctx.channel().remoteAddress());

        log.info("源客户端请求连接[{}:{}, channelId={}]", address.getHostString(), address.getPort(), channel.id());

        Client clientInfo = JSONObject.parseObject(new String(message.getData()), Client.class);

        String clientId = clientInfo.getId();

        //设置链路类型为客户端
        channel.attr(Constants.CLIENT_ID).set(clientId);

        //获取客户端信息
        Client client = ClientManager.getClient(clientId);
        if (client == null) {
            log.info("客户端[{}]不存在,连接关闭", clientId);

            message.setContent(JSONObject.toJSONString(new Message.Content(403, "客户端不存在, 请先申请!")));
            message.setType(Message.TYPE.UNLINK);

            channel.writeAndFlush(message);
            channel.close().sync();
        } else if (client.isOnline()) {
            if (clientInfo.getMacAddress().equals(client.getMacAddress())) {
                log.info("客户端[{}]重连", clientId);

                //关闭旧连接
                client.getChannel().close().sync();

                client.setChannel(channel);

                //返回连接成功消息
                message.setContent(JSONObject.toJSONString(new Message.Content(200, "客户端重连成功!")));
                message.setType(Message.TYPE.LINK);

                message.setData(JSONObject.toJSONString(client.copy()).getBytes(StandardCharsets.UTF_8));

                channel.writeAndFlush(message);

                handleSendTunnelMessage(channel, clientId);

                openPortMessage(channel, clientId);

                sendOpenPortMessage(channel, clientId);

                channel.config().setOption(ChannelOption.AUTO_READ, true);

            } else {
                //客户端id已被占用
                log.info("客户端[{}]已被占用,连接关闭", clientId);

                message.setContent(JSONObject.toJSONString(new Message.Content(403, "客户端已被占用, 请重新申请!")));
                message.setType(Message.TYPE.UNLINK);

                channel.writeAndFlush(message);
                channel.close().sync();

            }

        } else {
            client.setOnline(true);
            client.setChannel(channel);
            client.setHost(address.getHostString());

            //更新客户端
            ClientManager.updateClientMap(client);


            //返回连接成功消息
            message.setContent(JSONObject.toJSONString(new Message.Content(200, "客户端连接成功!")));
            message.setType(Message.TYPE.LINK);

            message.setData(JSONObject.toJSONString(client.copy()).getBytes(StandardCharsets.UTF_8));

            channel.writeAndFlush(message);

            handleSendTunnelMessage(channel, clientId);

            openPortMessage(channel, clientId);

            sendOpenPortMessage(channel, clientId);

            channel.config().setOption(ChannelOption.AUTO_READ, true);
        }
    }

    /*
     * 下发打开端口信息
     * */
    private void sendOpenPortMessage(Channel channel, String clientId) {

        //获取当前客户端是to的隧道, 表示要在客户端监听该端口
        Collection<Tunnel> byToClient = TunnelManager.getByToClient(clientId);
        byToClient.forEach(e -> {
            //如果隧道开启，检查对向客户端在线则向本客户端下发命令开启端口
            if (e.getState() == 1) {
                //获取对向客户端
                Client client = ClientManager.getClient(e.getFromClient());

                if (client != null) {
                    Tunnel tunnel = e.copy();
                    //检查对向客户端在线
                    if (client.isOnline()) {
                        /*
                         * 向当前客户端下发开启命令
                         */
                        log.info("向当前客户端[{}]下发开启端口命令[{}]", tunnel.getToClient(), tunnel.getToPort());

                        //下发数据
                        Message message = new Message();

                        message.setContent(Constants.STATE.REQUEST.value);
                        message.setData(JSONObject.toJSONString(tunnel).getBytes(StandardCharsets.UTF_8));
                        message.setType(Message.TYPE.OPEN_PORT);

                        channel.writeAndFlush(message);

                    }
                }

            }
        });
        //获取当前客户端是from的隧道, 检查对向客户端在线, 则向对向客户端发送打开端口命令
        Collection<Tunnel> byFromClient = TunnelManager.getByFromClient(clientId);
        byFromClient.forEach(e -> {
            //如果隧道开启，检查对向客户端在线则向两个客户端同时下发命令开启隧道
            if (e.getState() == 1) {
                //获取对向客户端
                Client client = ClientManager.getClient(e.getToClient());
                if (client != null && !client.getId().equals("server")) {
                    Tunnel tunnel = e.copy();
                    //客户端在线
                    if (client.isOnline()) {
                        /*
                         * 向对向客户端下发打开命令
                         */
                        log.info("向对向客户端[{}]下发开启端口命令[{}]", tunnel.getToClient(), tunnel.getToPort());

                        //下发数据
                        Message message = new Message();

                        message.setContent(Constants.STATE.REQUEST.value);
                        message.setData(JSONObject.toJSONString(tunnel).getBytes(StandardCharsets.UTF_8));
                        message.setType(Message.TYPE.OPEN_PORT);

                        client.getChannel().writeAndFlush(message);
                    }
                }

            }
        });

    }

    /*
     * 关闭端口信息
     * */
    private void handleClosePortMessage(ChannelHandlerContext ctx, Message message) {

    }

    /*
     * 处理打开端口信息
     * */
    private void handleOpenPortMessage(ChannelHandlerContext ctx, Message message) {
        Tunnel tunnel = JSONObject.parseObject(new String(message.getData()), Tunnel.class);

        if (message.getContent().equals(Constants.STATE.SUCCESS.value)) {
            log.info("客户端[{}]在隧道[{}]打开端口[{}]成功!", tunnel.getToClient(), tunnel.getId(), tunnel.getToPort());

            //检测客户端能否连接目标端口在更新隧道状态  TODO
            TunnelManager.getTunnel(tunnel.getId()).setAlive(true);

        } else if (message.getContent().equals(Constants.STATE.FAILED.value)) {
            log.info("客户端[{}]在隧道[{}]打开端口[{}]失败!", tunnel.getToClient(), tunnel.getId(), tunnel.getToPort());
        }

    }

    /*
     * 处理打开端口信息
     * */
    private void openPortMessage(Channel channel, String clientId) {
        Collection<Tunnel> tunnels = TunnelManager.getByFromClient(clientId);
        for (Tunnel tunnel : tunnels) {
            if (tunnel.getToClient().equals("server")) {
                try {
                    ChannelFuture future = UserServerBootstrap.start(tunnel.getToPort());
                    FutureManager.updateChannelFuture(tunnel.getId(), future);
                    log.info("服务端开启端口成功[port={}]", tunnel.getToPort());
                } catch (Exception e) {
                    log.info("服务端端口已开启[port={}], message: {}", tunnel.getToPort(), e.getMessage());
                }
            }
        }
    }

    private void handleSendTunnelMessage(Channel channel, String clientId) {

        //获取当前客户端关联的隧道, 下发隧道信息
        Collection<Tunnel> relevantTunnel = TunnelManager.getRelevantTunnelByClient(clientId);
        relevantTunnel.forEach(tunnel -> {
            log.info("向当前客户端[{}]下发隧道信息[{}]", clientId, tunnel.getId());

            //下发数据
            Message message = new Message();
            message.setContent(null);
            message.setData(JSONObject.toJSONString(tunnel).getBytes(StandardCharsets.UTF_8));
            message.setType(Message.TYPE.TUNNEL_MESSAGE);

            channel.writeAndFlush(message);

        });

    }

    private void handleOpenConnectMessage(ChannelHandlerContext ctx, Message message) throws InterruptedException {

        Channel channel = ctx.channel();
        Tunnel tunnel = JSONObject.parseObject(new String(message.getData()), Tunnel.class);

        Client toClient = ClientManager.getClient(tunnel.getToClient());
        Client fromClient = ClientManager.getClient(tunnel.getFromClient());

        //如果ToClient是服务端, 直接处理
        if (tunnel.getToClient().equals("server")) {
            //从ChannelManager中取出Channel
            Channel toChannel = ChannelManager.takeOutChannel(tunnel.getChannelId());

            if (message.getContent().equals(Constants.STATE.SUCCESS.value)) {
                log.info("链路开启成功, 此次在端口[{}]的请求转发开启可读, [tunnelId={},fromClient={}, toClient={}]", tunnel.getToPort(), tunnel.getId(), fromClient.getId(), toClient.getId());

                //关联隧道
                toChannel.attr(Constants.TOWARD_CHANNEL).set(channel);
                channel.attr(Constants.TOWARD_CHANNEL).set(toChannel);

                //设置可读
                toChannel.config().setOption(ChannelOption.AUTO_READ, true);

            } else if (message.getContent().equals(Constants.STATE.FAILED.value)) {
                log.info("链路开启失败, 取消此次在端口[{}]的请求转发, [tunnelId={},fromClient={}, toClient={}]", tunnel.getToPort(), tunnel.getId(), fromClient.getId(), toClient.getId());

                //关闭链路, 直接丢弃数据
                toChannel.close().sync();
            }
        } else {//如果ToClient不是服务端, 需要转发等处理

            if (message.getContent().equals(Constants.STATE.SUCCESS.value)) {
                log.info("链路开启成功, 回复原始请求客户端[{}], [tunnelId={},fromClient={}, toClient={}]", toClient.getId(), tunnel.getId(), fromClient.getId(), toClient.getId());
                //从ChannelManager中取出Channel
                Channel toChannel = ChannelManager.takeOutChannel(tunnel.getChannelId());
                //关联隧道
                toChannel.attr(Constants.TOWARD_CHANNEL).set(channel);
                channel.attr(Constants.TOWARD_CHANNEL).set(toChannel);

                //回复toChannel开启隧道成功消息
                toChannel.writeAndFlush(message);

            } else if (message.getContent().equals(Constants.STATE.FAILED.value)) {
                log.info("链路开启失败, 回复原始请求客户端[{}], [tunnelId={},fromClient={}, toClient={}]", toClient.getId(), tunnel.getId(), fromClient.getId(), toClient.getId());

                Channel toChannel = toClient.getChannel();
                //回复toChannel开启隧道失败消息
                toChannel.writeAndFlush(message);

            } else if (message.getContent().equals(Constants.STATE.REQUEST.value)) {
                if (tunnel.getFromClient().equals("server")) {
                    //请求服务端直接连接目标
                    ChannelFuture future = UserClientBootstrap.getUserBootstrap().connect(new InetSocketAddress(tunnel.getFromHost(), tunnel.getFromPort())).sync();
                    // 连接成功
                    if (future.isSuccess()) {
                        Channel toChannel = future.channel();
                        //标记此链路在服务器上
                        toChannel.attr(Constants.CHANNEL_ON_SERVER).set(true);

                        toChannel.attr(Constants.TOWARD_CHANNEL).set(channel);
                        channel.attr(Constants.TOWARD_CHANNEL).set(toChannel);

                        log.info("源[{}:{}]请求服务器连接目标[{}:{}]成功, 完成绑定双向链路!", tunnel.getToClient(), tunnel.getToPort(), tunnel.getFromHost(), tunnel.getFromPort());

                        message.setContent(Constants.STATE.SUCCESS.value);
                        //回复客户端成功消息, 使用新链路回复, 以便服务端绑定链路
                        channel.writeAndFlush(message);
                    } else {
                        log.info("源[{}:{}]请求连接目标[{}:{}]失败!", tunnel.getToClient(), tunnel.getToPort(), tunnel.getFromHost(), tunnel.getFromPort());

                        future.channel().close().sync();
                        channel.close().sync();

                        message.setContent(Constants.STATE.FAILED.value);
                        //回复客户端
                        channel.writeAndFlush(message);
                    }

                } else {
                    //要求某客户端连接目标
                    log.info("转发连接命令到目标客户端[{}], [tunnelId={},fromClient={}, toClient={}]", fromClient.getId(), tunnel.getId(), fromClient.getId(), toClient.getId());

                    //先把链路Channel保存起来
                    ChannelManager.updateChannel(channel);

                    //把channelId记录进隧道信息随网络转发
                    tunnel.setChannelId(channel.id().toString());
                    message.setData(JSONObject.toJSONString(tunnel).getBytes(StandardCharsets.UTF_8));

                    //转发连接命令到目标客户端fromClient
                    Channel fromChannel = ClientManager.getClient(tunnel.getFromClient()).getChannel();
                    fromChannel.writeAndFlush(message);
                }
            }
        }


    }

    private void handleCloseConnectMessage(ChannelHandlerContext ctx, Message message) throws InterruptedException {
        Channel channel = ctx.channel();
        Channel towardChannel = channel.attr(Constants.TOWARD_CHANNEL).get();
        log.info("关闭链路[({})<->({})]", channel.id(), towardChannel.id());
        //转发关闭消息
        towardChannel.close().sync();
        channel.close().sync();
    }

    private void handleHeartbeatMessage(ChannelHandlerContext ctx, Message message) {
        Message heartbeatMessage = new Message();
        heartbeatMessage.setType(Message.TYPE.HEARTBEAT);
        if (isDebugger) {
            log.info("回复心跳[{}]", ctx.channel().id());
        }
        ctx.channel().writeAndFlush(heartbeatMessage);
    }

    /*
     * 关闭连接信息
     * */
    private void sendCloseConnectMessage(Channel channel, Message message) {

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        String clientId = channel.attr(Constants.CLIENT_ID).get();
        InetSocketAddress address = (InetSocketAddress) (ctx.channel().remoteAddress());

        if (clientId != null) {
            //如果不空则为客户端链路
            Client client = ClientManager.getClient(clientId);
            client.setChannel(null);
            client.setOnline(false);

            Collection<Tunnel> tunnels = TunnelManager.getRelevantTunnelByClient(clientId);
            tunnels.forEach(tunnel -> {
                tunnel.setAlive(false);
                //下发对向客户端关闭监听端口
                if (tunnel.getFromClient().equals(clientId)) {
                    String toClient = tunnel.getToClient();
                    if (toClient.equals("server")) {
                        //如果对向客服端是服务器,直接停止监听端口
                        try {
                            FutureManager.close(tunnel.getId());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    } else {
                        Channel toChannel = ClientManager.getClient(toClient).getChannel();

                        Message message = new Message();
                        message.setType(Message.TYPE.CLOSE_PORT);
                        message.setData(JSONObject.toJSONString(tunnel).getBytes(StandardCharsets.UTF_8));

                        toChannel.writeAndFlush(message);
                    }
                }

            });
            channel.close().sync();
            log.info("客户端关闭[{}:{},clientId={},channelId={}], message: {}", address.getHostString(), address.getPort(), clientId, channel.id(), cause.getMessage());
        } else {
            //否则为隧道链路
            Channel towardChannel = channel.attr(Constants.TOWARD_CHANNEL).get();

            Message message = new Message();
            message.setType(Message.TYPE.CLOSE_CONNECT);

            //下发关闭对向连接链路
            if (towardChannel != null) {
                towardChannel.writeAndFlush(message);
                towardChannel.close().sync();
                channel.close().sync();
                log.info("链路关闭[{}:{},{}<->{}], message: {}", address.getHostString(), address.getPort(), channel.id(), towardChannel.id(), cause.getMessage());
            }
        }
    }
}